package com.smart.community.commons.websocket.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smart.community.commons.websocket.WebSocketMessage;
import com.smart.community.commons.websocket.WebSocketService;
import com.smart.community.commons.websocket.UserWebSocketSession;
import com.smart.community.commons.websocket.SimpleWebSocketMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * WebSocket服务实现类
 * 
 * @author Wu.Liang
 * @since 2025-01-30
 */
@Service
@Slf4j
public class WebSocketServiceImpl implements WebSocketService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private final ObjectMapper objectMapper = new ObjectMapper();

    // 内存缓存：用户会话映射
    private final Map<Long, UserWebSocketSession> userSessions = new ConcurrentHashMap<>();
    
    // 内存缓存：房间用户映射
    private final Map<String, Set<Long>> roomUsers = new ConcurrentHashMap<>();
    
    // 内存缓存：房间信息
    private final Map<String, RoomInfo> roomInfo = new ConcurrentHashMap<>();
    
    // 内存缓存：WebSocket连接映射
    private final Map<Long, org.springframework.web.socket.WebSocketSession> webSocketConnections = new ConcurrentHashMap<>();
    
    // 消息重试队列
    private final Map<String, RetryMessage> retryQueue = new ConcurrentHashMap<>();
    
    // 定时任务执行器
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

    // Redis键前缀
    private static final String WEBSOCKET_SESSION_PREFIX = "websocket:session:";
    private static final String WEBSOCKET_ROOM_PREFIX = "websocket:room:";
    private static final String WEBSOCKET_MESSAGE_PREFIX = "websocket:message:";
    private static final String WEBSOCKET_STATS_PREFIX = "websocket:stats:";

    // 配置常量
    private static final int MAX_RETRY_COUNT = 3;
    private static final long RETRY_DELAY_MS = 5000; // 5秒
    private static final int SESSION_TIMEOUT_MINUTES = 30;
    private static final int HEARTBEAT_INTERVAL_SECONDS = 30;

    /**
     * 初始化定时任务
     */
    public WebSocketServiceImpl() {
        // 启动心跳检测
        scheduler.scheduleAtFixedRate(this::heartbeatCheck, 0, HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS);
        
        // 启动会话清理
        scheduler.scheduleAtFixedRate(this::cleanupExpiredSessions, 1, 5, TimeUnit.MINUTES);
        
        // 启动消息重试处理
        scheduler.scheduleAtFixedRate(this::processRetryQueue, 0, 10, TimeUnit.SECONDS);
        
        log.info("WebSocket服务初始化完成");
    }

    @Override
    public boolean sendToUser(Long userId, WebSocketMessage message) {
        try {
            UserWebSocketSession session = userSessions.get(userId);
            if (session != null && UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus())) {
                // 更新消息状态
                message.setStatus(WebSocketMessage.MessageStatus.SENT);
                message.setTimestamp(LocalDateTime.now());
                
                // 获取WebSocket连接
                org.springframework.web.socket.WebSocketSession wsSession = webSocketConnections.get(userId);
                if (wsSession != null && wsSession.isOpen()) {
                    // 发送消息
                    String messageJson = objectMapper.writeValueAsString(message);
                    wsSession.sendMessage(new TextMessage(messageJson));
                    
                    // 更新用户活跃时间
                    session.updateLastActiveTime();
                    
                    // 保存到Redis
                    saveMessageToRedis(userId, message);
                    
                    // 更新统计信息
                    updateMessageStats(userId, true);
                    
                    log.debug("消息发送成功，用户ID: {}, 消息类型: {}", userId, message.getMessageType());
                    return true;
                } else {
                    // 连接不存在或已关闭，加入重试队列
                    addToRetryQueue(userId, message);
                    return false;
                }
            } else {
                log.warn("用户不在线，用户ID: {}", userId);
                return false;
            }
        } catch (Exception e) {
            log.error("发送消息失败，用户ID: {}, 错误: {}", userId, e.getMessage(), e);
            
            // 加入重试队列
            addToRetryQueue(userId, message);
            return false;
        }
    }

    @Override
    public int sendToRoom(String roomId, WebSocketMessage message) {
        Set<Long> userIds = roomUsers.get(roomId);
        if (userIds == null || userIds.isEmpty()) {
            log.warn("房间内没有用户，房间ID: {}", roomId);
            return 0;
        }

        int successCount = 0;
        for (Long userId : userIds) {
            if (sendToUser(userId, message)) {
                successCount++;
            }
        }

        log.info("房间消息发送完成，房间ID: {}, 成功发送: {}/{}", roomId, successCount, userIds.size());
        return successCount;
    }

    @Override
    public int broadcast(WebSocketMessage message) {
        List<UserWebSocketSession> onlineUsers = getOnlineUsers();
        int successCount = 0;

        for (UserWebSocketSession session : onlineUsers) {
            if (sendToUser(session.getUserId(), message)) {
                successCount++;
            }
        }

        log.info("广播消息发送完成，成功发送: {}/{}", successCount, onlineUsers.size());
        return successCount;
    }

    @Override
    public int broadcastToPropertyCompany(Long propertyCompanyId, WebSocketMessage message) {
        List<UserWebSocketSession> companyUsers = getPropertyCompanyUsers(propertyCompanyId);
        int successCount = 0;

        for (UserWebSocketSession session : companyUsers) {
            if (sendToUser(session.getUserId(), message)) {
                successCount++;
            }
        }

        log.info("物业公司广播消息发送完成，物业公司ID: {}, 成功发送: {}/{}", 
            propertyCompanyId, successCount, companyUsers.size());
        return successCount;
    }

    @Override
    public List<UserWebSocketSession> getOnlineUsers() {
        return userSessions.values().stream()
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .collect(Collectors.toList());
    }

    @Override
    public List<UserWebSocketSession> getRoomUsers(String roomId) {
        Set<Long> userIds = roomUsers.get(roomId);
        if (userIds == null) {
            return new ArrayList<>();
        }

        return userIds.stream()
                .map(userSessions::get)
                .filter(Objects::nonNull)
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .collect(Collectors.toList());
    }

    @Override
    public List<UserWebSocketSession> getPropertyCompanyUsers(Long propertyCompanyId) {
        return userSessions.values().stream()
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .filter(session -> Objects.equals(session.getPropertyCompanyId(), propertyCompanyId))
                .collect(Collectors.toList());
    }

    @Override
    public boolean isUserOnline(Long userId) {
        UserWebSocketSession session = userSessions.get(userId);
        return session != null && 
               UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()) && 
               !session.isExpired();
    }

    @Override
    public UserWebSocketSession getUserSession(Long userId) {
        return userSessions.get(userId);
    }

    @Override
    public boolean createRoom(String roomId, String roomName, String description) {
        if (roomInfo.containsKey(roomId)) {
            log.warn("房间已存在，房间ID: {}", roomId);
            return false;
        }

        RoomInfo room = new RoomInfo();
        room.setRoomId(roomId);
        room.setRoomName(roomName);
        room.setDescription(description);
        room.setCreateTime(LocalDateTime.now());
        room.setUserCount(0);

        roomInfo.put(roomId, room);
        roomUsers.put(roomId, ConcurrentHashMap.newKeySet());

        // 保存到Redis
        String roomKey = WEBSOCKET_ROOM_PREFIX + roomId;
        redisTemplate.opsForValue().set(roomKey, room);

        log.info("房间创建成功，房间ID: {}, 房间名称: {}", roomId, roomName);
        return true;
    }

    @Override
    public boolean deleteRoom(String roomId) {
        RoomInfo room = roomInfo.remove(roomId);
        Set<Long> userIds = roomUsers.remove(roomId);

        if (room != null) {
            // 通知房间内所有用户离开
            if (userIds != null) {
                for (Long userId : userIds) {
                    UserWebSocketSession session = userSessions.get(userId);
                    if (session != null) {
                        session.unsubscribeRoom(roomId);
                    }
                }
            }

            // 从Redis中删除
            String roomKey = WEBSOCKET_ROOM_PREFIX + roomId;
            redisTemplate.delete(roomKey);

            log.info("房间删除成功，房间ID: {}", roomId);
            return true;
        }

        return false;
    }

    @Override
    public Set<String> getRoomList() {
        return new HashSet<>(roomInfo.keySet());
    }

    @Override
    public boolean joinRoom(Long userId, String roomId) {
        if (!roomInfo.containsKey(roomId)) {
            log.warn("房间不存在，房间ID: {}", roomId);
            return false;
        }

        UserWebSocketSession session = userSessions.get(userId);
        if (session == null) {
            log.warn("用户会话不存在，用户ID: {}", userId);
            return false;
        }

        Set<Long> userIds = roomUsers.computeIfAbsent(roomId, k -> ConcurrentHashMap.newKeySet());
        userIds.add(userId);

        session.subscribeRoom(roomId, roomInfo.get(roomId).getRoomName());

        // 更新房间用户数量
        RoomInfo room = roomInfo.get(roomId);
        room.setUserCount(userIds.size());

        log.info("用户加入房间成功，用户ID: {}, 房间ID: {}", userId, roomId);
        return true;
    }

    @Override
    public boolean leaveRoom(Long userId, String roomId) {
        Set<Long> userIds = roomUsers.get(roomId);
        if (userIds == null) {
            return false;
        }

        boolean removed = userIds.remove(userId);
        if (removed) {
            UserWebSocketSession session = userSessions.get(userId);
            if (session != null) {
                session.unsubscribeRoom(roomId);
            }

            // 更新房间用户数量
            RoomInfo room = roomInfo.get(roomId);
            if (room != null) {
                room.setUserCount(userIds.size());
            }

            log.info("用户离开房间成功，用户ID: {}, 房间ID: {}", userId, roomId);
        }

        return removed;
    }

    @Override
    public int cleanupExpiredSessions() {
        int cleanedCount = 0;
        LocalDateTime now = LocalDateTime.now();

        // 清理过期会话
        Iterator<Map.Entry<Long, UserWebSocketSession>> iterator = userSessions.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Long, UserWebSocketSession> entry = iterator.next();
            UserWebSocketSession session = entry.getValue();

            if (session.isExpired()) {
                // 从所有房间中移除用户
                for (String roomId : session.getSubscribedRooms().keySet()) {
                    leaveRoom(session.getUserId(), roomId);
                }

                // 关闭WebSocket连接
                org.springframework.web.socket.WebSocketSession wsSession = webSocketConnections.remove(session.getUserId());
                if (wsSession != null && wsSession.isOpen()) {
                    try {
                        wsSession.close();
                    } catch (IOException e) {
                        log.warn("关闭WebSocket连接失败，用户ID: {}, 错误: {}", session.getUserId(), e.getMessage());
                    }
                }

                // 设置会话状态为断开
                session.setStatus(UserWebSocketSession.SessionStatus.DISCONNECTED);

                iterator.remove();
                cleanedCount++;

                log.info("清理过期会话，用户ID: {}, 用户名: {}", session.getUserId(), session.getUsername());
            }
        }

        log.info("会话清理完成，清理数量: {}", cleanedCount);
        return cleanedCount;
    }

    @Override
    public int getOnlineUserCount() {
        return (int) userSessions.values().stream()
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .count();
    }

    @Override
    public int getRoomUserCount(String roomId) {
        Set<Long> userIds = roomUsers.get(roomId);
        if (userIds == null) {
            return 0;
        }

        return (int) userIds.stream()
                .map(userSessions::get)
                .filter(Objects::nonNull)
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .count();
    }

    @Override
    public int getPropertyCompanyUserCount(Long propertyCompanyId) {
        return (int) userSessions.values().stream()
                .filter(session -> UserWebSocketSession.SessionStatus.CONNECTED.equals(session.getStatus()))
                .filter(session -> !session.isExpired())
                .filter(session -> Objects.equals(session.getPropertyCompanyId(), propertyCompanyId))
                .count();
    }

    /**
     * 注册用户会话
     * 
     * @param userId 用户ID
     * @param wsSession WebSocket连接
     * @param session 用户会话信息
     */
    public void registerUserSession(Long userId, org.springframework.web.socket.WebSocketSession wsSession, UserWebSocketSession session) {
        userSessions.put(userId, session);
        webSocketConnections.put(userId, wsSession);
        
        // 保存到Redis
        String sessionKey = WEBSOCKET_SESSION_PREFIX + userId;
        redisTemplate.opsForValue().set(sessionKey, session, java.time.Duration.ofMinutes(SESSION_TIMEOUT_MINUTES));
        
        log.info("用户会话注册成功，用户ID: {}, 用户名: {}", userId, session.getUsername());
    }

    /**
     * 注销用户会话
     * 
     * @param userId 用户ID
     */
    public void unregisterUserSession(Long userId) {
        UserWebSocketSession session = userSessions.remove(userId);
        webSocketConnections.remove(userId);
        
        if (session != null) {
            // 从所有房间中移除用户
            for (String roomId : session.getSubscribedRooms().keySet()) {
                leaveRoom(userId, roomId);
            }
            
            // 从Redis中删除
            String sessionKey = WEBSOCKET_SESSION_PREFIX + userId;
            redisTemplate.delete(sessionKey);
            
            log.info("用户会话注销成功，用户ID: {}, 用户名: {}", userId, session.getUsername());
        }
    }

    /**
     * 心跳检测
     */
    private void heartbeatCheck() {
        try {
            List<UserWebSocketSession> onlineUsers = getOnlineUsers();
            for (UserWebSocketSession session : onlineUsers) {
                // 发送心跳消息
                SimpleWebSocketMessage heartbeat = new SimpleWebSocketMessage("HEARTBEAT", "ping");
                heartbeat.setSenderId(0L); // 系统消息
                heartbeat.setSenderName("SYSTEM");
                
                sendToUser(session.getUserId(), heartbeat);
            }
        } catch (Exception e) {
            log.error("心跳检测失败，错误: {}", e.getMessage(), e);
        }
    }

    /**
     * 处理重试队列
     */
    private void processRetryQueue() {
        Iterator<Map.Entry<String, RetryMessage>> iterator = retryQueue.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, RetryMessage> entry = iterator.next();
            RetryMessage retryMessage = entry.getValue();
            
            if (retryMessage.shouldRetry()) {
                // 尝试重新发送
                if (sendToUser(retryMessage.getUserId(), retryMessage.getMessage())) {
                    iterator.remove();
                    log.debug("重试消息发送成功，用户ID: {}, 消息ID: {}", 
                        retryMessage.getUserId(), retryMessage.getMessage().getMessageId());
                } else {
                    retryMessage.incrementRetryCount();
                    if (retryMessage.getRetryCount() >= MAX_RETRY_COUNT) {
                        iterator.remove();
                        log.warn("消息重试次数超限，放弃发送，用户ID: {}, 消息ID: {}", 
                            retryMessage.getUserId(), retryMessage.getMessage().getMessageId());
                    }
                }
            } else {
                iterator.remove();
            }
        }
    }

    /**
     * 添加消息到重试队列
     */
    private void addToRetryQueue(Long userId, WebSocketMessage message) {
        String retryKey = userId + ":" + message.getMessageId();
        RetryMessage retryMessage = new RetryMessage(userId, message);
        retryQueue.put(retryKey, retryMessage);
        
        log.debug("消息加入重试队列，用户ID: {}, 消息ID: {}", userId, message.getMessageId());
    }

    /**
     * 保存消息到Redis
     */
    private void saveMessageToRedis(Long userId, WebSocketMessage message) {
        try {
            String messageKey = WEBSOCKET_MESSAGE_PREFIX + userId + ":" + message.getMessageId();
            redisTemplate.opsForValue().set(messageKey, message, java.time.Duration.ofHours(24));
        } catch (Exception e) {
            log.warn("保存消息到Redis失败，用户ID: {}, 错误: {}", userId, e.getMessage());
        }
    }

    /**
     * 更新消息统计信息
     */
    private void updateMessageStats(Long userId, boolean success) {
        try {
            String statsKey = WEBSOCKET_STATS_PREFIX + userId;
            Map<String, Object> stats = (Map<String, Object>) redisTemplate.opsForValue().get(statsKey);
            
            if (stats == null) {
                stats = new HashMap<>();
            }
            
            // 更新发送统计
            Long totalSent = (Long) stats.getOrDefault("totalSent", 0L);
            stats.put("totalSent", totalSent + 1);
            
            if (success) {
                Long successSent = (Long) stats.getOrDefault("successSent", 0L);
                stats.put("successSent", successSent + 1);
            }
            
            stats.put("lastUpdateTime", LocalDateTime.now());
            
            redisTemplate.opsForValue().set(statsKey, stats, java.time.Duration.ofDays(7));
        } catch (Exception e) {
            log.warn("更新消息统计失败，用户ID: {}, 错误: {}", userId, e.getMessage());
        }
    }

    /**
     * 获取用户消息统计
     */
    public Map<String, Object> getUserMessageStats(Long userId) {
        try {
            String statsKey = WEBSOCKET_STATS_PREFIX + userId;
            Map<String, Object> stats = (Map<String, Object>) redisTemplate.opsForValue().get(statsKey);
            
            if (stats == null) {
                stats = new HashMap<>();
                stats.put("totalSent", 0L);
                stats.put("successSent", 0L);
                stats.put("lastUpdateTime", LocalDateTime.now());
            }
            
            return stats;
        } catch (Exception e) {
            log.warn("获取用户消息统计失败，用户ID: {}, 错误: {}", userId, e.getMessage());
            return new HashMap<>();
        }
    }

    /**
     * 房间信息内部类
     */
    @lombok.Data
    private static class RoomInfo {
        private String roomId;
        private String roomName;
        private String description;
        private LocalDateTime createTime;
        private int userCount;
    }

    /**
     * 重试消息内部类
     */
    @lombok.Data
    private static class RetryMessage {
        private Long userId;
        private WebSocketMessage message;
        private int retryCount;
        private LocalDateTime lastRetryTime;

        public RetryMessage(Long userId, WebSocketMessage message) {
            this.userId = userId;
            this.message = message;
            this.retryCount = 0;
            this.lastRetryTime = LocalDateTime.now();
        }

        public boolean shouldRetry() {
            return LocalDateTime.now().isAfter(lastRetryTime.plus(java.time.Duration.ofMillis(RETRY_DELAY_MS)));
        }

        public void incrementRetryCount() {
            this.retryCount++;
            this.lastRetryTime = LocalDateTime.now();
        }
    }
}
